package woaini.wind.bot.chat.stream;

import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Flow;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 聊天流式对话 获取到对接流之后变量处理
 *
 * @see WindChatStream
 * @author yefeng {@date 2025-06-20 15:49:14}
 */
@Slf4j
public class WindChatStream implements Flow.Subscriber<String> {

  private AtomicBoolean hasNext;

  private BlockingQueue<String> messages;

  private Flow.Subscription subscription;

  public WindChatStream() {
    this.hasNext = new AtomicBoolean(true);
    this.messages = new LinkedBlockingQueue<>(20);
  }

  public boolean hasNext() {
    return hasNext.get();
  }

  @Override
  public void onSubscribe(Flow.Subscription subscription) {
    log.info("WindChatStream: 开始订阅");
    this.subscription = subscription;
    subscription.request(1);
  }

  @Override
  public void onNext(String s) {
    log.info("WindChatStream-消息片段接收: {}", s);
    messages.add(s);
    subscription.request(1);
  }

  @Override
  public void onError(Throwable throwable) {
    log.error("WindChatStream: 错误: {}", throwable.getMessage());
    this.hasNext.set(false);
  }

  @Override
  public void onComplete() {
    this.hasNext.set(false);
  }

  public String getMessage() {
    try {
      String poll = messages.poll(5, TimeUnit.SECONDS);
      if (StrUtil.isEmpty(poll)){
        return null;
      }
      return poll;
    } catch (InterruptedException e) {
      this.hasNext.set(false);
      return null;
    }
  }
}
